package org.cy.spring.learning.webchat.websocket;

import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.connection.Message;
import org.springframework.data.redis.connection.MessageListener;
import org.springframework.data.redis.core.StringRedisTemplate;
import org.springframework.data.redis.listener.ChannelTopic;
import org.springframework.data.redis.listener.RedisMessageListenerContainer;
import org.springframework.data.redis.listener.Topic;
import org.springframework.data.redis.serializer.RedisSerializer;
import org.springframework.data.redis.serializer.StringRedisSerializer;
import org.springframework.web.socket.*;

import java.io.IOException;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * create date: 2019/9/26 19:01
 *
 * @author cao.yong
 */
@Slf4j
public class WebChatHandler implements WebSocketHandler, MessageListener {
    private ConcurrentMap<String, WebSocketSession> sessions = new ConcurrentHashMap<>(0);
    private ConcurrentMap<String, AtomicInteger> onlineRoomUserNum = new ConcurrentHashMap<>(0);

    @Autowired
    private RedisMessageListenerContainer container;

    @Autowired
    private StringRedisTemplate stringRedisTemplate;

    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        String roomId = (String) session.getAttributes().get("roomId");
        String userId = (String) session.getAttributes().get("userId");
        sessions.put(userId, session);
        if (onlineRoomUserNum.containsKey(roomId)) {
            AtomicInteger roomOnlineNum = onlineRoomUserNum.get(roomId);
            roomOnlineNum.incrementAndGet();
        } else {
            onlineRoomUserNum.put(roomId, new AtomicInteger(1));
        }
        container.addMessageListener(this, new ChannelTopic("room:" + roomId));
        for (ConcurrentMap.Entry<String, AtomicInteger> item : onlineRoomUserNum.entrySet()) {
            log.info("房间: {} , 在线人数: {}", item.getKey(), item.getValue().get());
        }
    }

    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus closeStatus) throws Exception {
        String roomId = (String) session.getAttributes().get("roomId");
        String userId = (String) session.getAttributes().get("userId");
        sessions.remove(userId);
        if (onlineRoomUserNum.containsKey(roomId)) {
            AtomicInteger roomOnlineNum = onlineRoomUserNum.get(roomId);
            roomOnlineNum.decrementAndGet();
        }
        container.removeMessageListener(this, new ChannelTopic("room:" + roomId));
        for (ConcurrentMap.Entry<String, AtomicInteger> item : onlineRoomUserNum.entrySet()) {
            log.info("房间: {} , 在线人数: {}", item.getKey(), item.getValue().get());
        }
    }

    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws Exception {
        String roomId = (String) session.getAttributes().get("roomId");
        String userId = (String) session.getAttributes().get("userId");
        TextMessage msg = (TextMessage) message;
        log.info("收到用户[{}], 消息: {}", userId, msg.getPayload());
        stringRedisTemplate.convertAndSend("room:" + roomId, userId + " 说: " + msg.getPayload());

    }

    @Override
    public void handleTransportError(WebSocketSession session, Throwable throwable) throws Exception {

    }

    @Override
    public boolean supportsPartialMessages() {
        return false;
    }

    @Override
    public void onMessage(Message message, byte[] bytes) {
        String msg = new StringRedisSerializer().deserialize(message.getBody());
        log.info("收到订阅消息: {}", msg);
        for (ConcurrentMap.Entry<String, WebSocketSession> item : sessions.entrySet()) {
            TextMessage textMessage = new TextMessage(msg);
            try {
                item.getValue().sendMessage(textMessage);
            } catch (IOException e) {
                e.printStackTrace();
            }
        }
    }
}
